package com.wei.violet.app.Label

import com.wei.violet.utils.Jpools
import org.apache.commons.lang.StringUtils
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext}

object LabelHandler {


  Logger.getLogger("org").setLevel(Level.WARN)

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setAppName("将终端设备统计结果添加到mysql数据库中")
      .setMaster("local[*]")
      .set("spark.serizlizer", "org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(conf)
    val sQLContext = new SQLContext(sc)
    val dataFrame: DataFrame = sQLContext.read.parquet("F:\\bigdataFile\\parquet")

    val data = dataFrame.mapPartitions(m => {
      var Labelmap = Map[String, Int]()
      val jedis = Jpools.getJedis
      val labeldata = m.map(data => {
        val adspacetype = data.getAs[Int]("adspacetype")
        val adspacetypename = data.getAs[String]("adspacetypename")
        val appname = data.getAs[String]("appname")
        val appid = data.getAs[String]("appid")
        val adplatformproviderid = data.getAs[Int]("adplatformproviderid")
        val client = data.getAs[Int]("client") //操作系统

        val ispid = data.getAs[Int]("ispid") //运营商

        val networkmannerid = data.getAs[Int]("networkmannerid") //联网方式

        val keywords = data.getAs[String]("keywords") //关键字
        val provincename = data.getAs[String]("provincename")
        val cityname = data.getAs[String]("cityname")

        val keywordssplit = keywords.split("\\|").filter(x => (x.length >= 3 && x.length <= 8))
        //广告位类型标签
        /*val adspacetypeLabel */
        if (adspacetype < 10)
          Labelmap += ("LC" + "0" + adspacetype -> 1)
        else
          Labelmap += ("LC" + adspacetype -> 1)

        Labelmap += ("LN" + adspacetypename -> 1) //广告位类型名称标签

        if (StringUtils.isEmpty(appname)) {
          if (StringUtils.isNotEmpty(appid)) {
            val result = jedis.hget("appdict", appid)
            if (StringUtils.isNotEmpty(result))
              Labelmap += ("APP" + result -> 1)

          }
        } else {

          Labelmap += ("APP" + appname -> 1)
        }
        Labelmap += ("CN" + adplatformproviderid -> 1) //渠道标签

        if (client == 1) {
          Labelmap += ("D00010001" -> 1)
        } else if (client == 2) {
          Labelmap += ("D00010002" -> 1)
        } else if (client == 3) {
          Labelmap += ("D00010003" -> 1)
        } else {
          Labelmap += ("D00010004" -> 1)
        }
        if (networkmannerid == 1) {
          Labelmap += ("D00020004" -> 1)
        } else if (networkmannerid == 2) {
          Labelmap += ("D00020003" -> 1)
        } else if (networkmannerid == 3) {
          Labelmap += ("D00020001" -> 1)
        } else if (networkmannerid == 5) {
          Labelmap += ("D00020002" -> 1)
        } else {
          Labelmap += ("D00020005" -> 1)
        }
        val ispLabel = if (ispid == 1) {
          Labelmap += ("D00030001" -> 1)
        } else if (ispid == 2) {
          Labelmap += ("D00030002" -> 1)
        } else if (ispid == 3) {
          Labelmap += ("D00030003" -> 1)
        } else {
          Labelmap += ("D00030004" -> 1)
        }
        val keywordsLabel =
          keywordssplit.map(x => ("K".concat(x.replace(":", ""))) -> 1).toMap[String, Int]

        Labelmap += ("ZP" + provincename -> 1)
        Labelmap += ("ZC" + cityname -> 1)
        //取用户id
        val imei = data.getAs[String]("imei")
        val mac = data.getAs[String]("mac")
        val idfa = data.getAs[String]("idfa")
        val openudid = data.getAs[String]("openudid")
        val androidid = data.getAs[String]("androidid")
        val imeimd5 = data.getAs[String]("imeimd5")
        val macmd5 = data.getAs[String]("macmd5")
        val idfamd5 = data.getAs[String]("idfamd5")
        val openudidmd5 = data.getAs[String]("openudidmd5")
        val androididmd5 = data.getAs[String]("androididmd5")
        val imeisha1 = data.getAs[String]("imeisha1")
        val macsha1 = data.getAs[String]("macsha1")
        val idfasha1 = data.getAs[String]("idfasha1")
        val openudidsha1 = data.getAs[String]("openudidsha1")
        val androididsha1 = data.getAs[String]("androididsha1")
        var userId = ""

        if (StringUtils.isNotEmpty(imei)) {
          userId = imei
        }
        else if (StringUtils.isNotEmpty(mac)) {
          userId = mac
        } else if (StringUtils.isNotEmpty(idfa)) {
          userId = idfa

        } else if (StringUtils.isNotEmpty(openudid)) {
          userId = openudid

        } else if (StringUtils.isNotEmpty(androidid)) {
          userId = androidid

        } else if (StringUtils.isNotEmpty(imeimd5)) {
          userId = imeimd5

        } else if (StringUtils.isNotEmpty(macmd5)) {
          userId = macmd5

        } else if (StringUtils.isNotEmpty(idfamd5)) {
          userId = idfamd5

        } else if (StringUtils.isNotEmpty(openudidmd5)) {
          userId = openudidmd5

        } else if (StringUtils.isNotEmpty(androididmd5)) {
          userId = androididmd5

        } else if (StringUtils.isNotEmpty(imeisha1)) {
          userId = imeisha1

        } else if (StringUtils.isNotEmpty(macsha1)) {
          userId = macsha1

        } else if (StringUtils.isNotEmpty(idfasha1)) {
          userId = idfasha1

        } else if (StringUtils.isNotEmpty(openudidsha1)) {
          userId = openudidsha1

        } else if (StringUtils.isNotEmpty(androididsha1)) {
          userId = androididsha1

        }

        (userId.toString, (keywordsLabel ++ Labelmap).toList)

      })

      jedis.close()
      labeldata

    }).filter(_._1.length > 0).reduceByKey {
      case (list1, list2) => {

        (list1 ++ list2)
          .groupBy(_._1)
          .map {
            case (k, list) => {
              (k, list.map(t => t._2).sum)
            }
          }.toList
      }
    }
    val dataOutputPath = "F:\\bigdataFile\\labeldata"
    val fs: FileSystem = FileSystem.get(sc.hadoopConfiguration)
    val path = new Path(dataOutputPath)
    if (fs.exists(path)) {
      fs.delete(path, true)
    }

    data.map(x => {
      //x._1 + "\t" + x._2.mkString("\t")
      x._1 + "\t" + x._2.map(a=>(a._1+":"+a._2)).mkString("\t")
    }).coalesce(6).saveAsTextFile(dataOutputPath)
    sc.stop()

  }


}
